package br.biofoco.p2p.utils;

import java.util.Iterator;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ConcurrentCache<K, V> {

	private final ConcurrentMap<K, CacheEntry<V>> map = new ConcurrentHashMap<K, CacheEntry<V>>();

	private CachePredicate<K,V> predicate;
	private final long scanInterval;
	private final long timeToLive;
	private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();

	public ConcurrentCache(long scanInterval, long timeToLive) {
		this.scanInterval = scanInterval;
		this.timeToLive = timeToLive;
	}

	public CacheEntry<V> put(K key, V value) {
		return map.put(key, new CacheEntry<V>(value, System.currentTimeMillis()));
	}

	public void setPredicate(CachePredicate<K, V> cachePredicate) {
		this.predicate = cachePredicate;
	}

	public void start() {
		executor.scheduleAtFixedRate(new EvictionThread(), scanInterval, scanInterval, TimeUnit.MILLISECONDS);
	}

	public void stop() {
		executor.shutdownNow();
	}

	private class EvictionThread implements Runnable {
		
		@Override
		public void run() {
			System.out.println("running cache ");
			System.out.println(map);
			doEviction();
		}

		private void doEviction() {
			for (Iterator<Entry<K, CacheEntry<V>>> it = map.entrySet().iterator(); it.hasNext();) {
				Entry<K, CacheEntry<V>> entry = it.next();
				if (predicate != null && predicate.apply(entry.getKey(), entry.getValue())) {
					System.out.println("removing key " + entry.getKey());
					it.remove();
					continue;
				}
				
				// bypass TTL
				if (timeToLive == 0){
					continue;
				}
								
				long currTime = System.currentTimeMillis();
				long entryTime= entry.getValue().getTimestamp() + timeToLive;
				System.out.println("currTime: " + currTime + " entryTime: " + entryTime);
				if (currTime >= entryTime) {
					System.out.println("expired key " + entry.getKey());
					it.remove();
				}
			}
		}
	}
	
	public static interface CachePredicate<K,V> {
		public boolean apply(K key, CacheEntry<V> entry);
	}
	
	public static class CacheEntry<V> {
		
		private final V value;
		private final long timestamp;
		
		public CacheEntry(V value, long timestamp) {
			this.value = value;
			this.timestamp = timestamp;
		}

		public V getValue() {
			return value;
		}

		public long getTimestamp() {
			return timestamp;
		}
		
		@Override
		public String toString() {
			return value + ":" + timestamp;
		}
	}
}